package com.cctc.rds.scheduler.server.raft.rpc.impl;

import com.cctc.rds.nrpc.core.util.JsonUtils;
import com.cctc.rds.raft.core.client.command.RaftCommandResponse;
import com.cctc.rds.scheduler.core.raft.TaskRpc;
import com.cctc.rds.scheduler.core.task.TaskLog;
import com.cctc.rds.scheduler.core.task.TaskStatus;
import com.cctc.rds.scheduler.server.raft.Command;
import com.cctc.rds.scheduler.server.raft.helper.RaftCommandClientHolder;
import com.cctc.rds.scheduler.server.raft.helper.RaftHelper;
import com.cctc.rds.scheduler.server.schedule.TaskScheduleServer;
import com.cctc.rds.scheduler.server.storage.Storage;
import com.cctc.rds.scheduler.server.util.TaskIdGenerator;

import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 任务实例日志 RPC 服务端 [响应] 实现类。
 * 主流程：存储（save）-> 取消（cancel）-> 拉取执行（pull）-> 提交执行结果（commit）。
 */
public class TaskRpcResponseImpl implements TaskRpc {
    private final Storage storage;
    private final TaskScheduleServer delayScheduleServer;

    public TaskRpcResponseImpl(Storage storage) {
        this.storage = storage;
        this.delayScheduleServer = new TaskScheduleServer();
    }

    @Override
    public long save(String app, TaskLog taskLog) {
        long taskId = TaskIdGenerator.genTaskId();
        taskLog.setTaskId(taskId);
        taskLog.setExecPriodIndex(0);
        taskLog.setStatus(TaskStatus.PENDING.getStatus());
        String command = Command.SAVE
                .replace("{app}", app)
                .replace("{taskId}", String.valueOf(taskId))
                .replace("{taskLog}", JsonUtils.toJson(taskLog));
        RaftCommandResponse response = RaftCommandClientHolder.getRaftCommandClient().exec(command.getBytes(StandardCharsets.UTF_8));
        if (response.isSuccess()) return taskId;
        throw new RuntimeException(JsonUtils.toJson(response));
    }

    @Override
    public boolean cancel(String app, long taskId) {
        TaskLog taskLog = storage.get(app, taskId);
        if (taskLog == null) return false;
        if (!taskLog.getStatus().equals(TaskStatus.PENDING.getStatus())) return false;
        long deadline = taskLog.getScheduleTime() + taskLog.getExecPriod().get(0);
        if (deadline - System.currentTimeMillis() / 1000 <= 0) return false;
        taskLog.setStatus(TaskStatus.CANCEL.getStatus());
        String command = Command.SAVE
                .replace("{app}", app)
                .replace("{taskId}", String.valueOf(taskId))
                .replace("{taskLog}", JsonUtils.toJson(taskLog));
        RaftCommandResponse response = RaftCommandClientHolder.getRaftCommandClient().exec(command.getBytes(StandardCharsets.UTF_8));
        if (response.isSuccess()) return true;
        throw new RuntimeException(JsonUtils.toJson(response));
    }

    @Override
    public List<TaskLog> pull(String app, int maxRecord) {
        RaftHelper.checkLeader();
        synchronized (this) {
            List<Long> taskIds = delayScheduleServer.pull(app, maxRecord);
            if (taskIds.isEmpty()) return Collections.emptyList();
            return taskIds.parallelStream().map(taskId -> storage.get(app, taskId)).collect(Collectors.toList());
        }
    }

    @Override
    public void commit(String app, long taskId, boolean success) {
        TaskLog taskLog = storage.get(app, taskId);
        if (taskLog == null) return;
        if (success) {
            taskLog.setStatus(TaskStatus.SUCCESS.getStatus());
        } else {
            taskLog.setRetryScheduleTime(System.currentTimeMillis() / 1000);
            taskLog.setExecPriodIndex(taskLog.getExecPriodIndex() + 1);
            if (taskLog.getExecPriodIndex() >= taskLog.getExecPriod().size()) {
                taskLog.setStatus(TaskStatus.FAIL.getStatus());
            } else {
                taskLog.setStatus(TaskStatus.PENDING.getStatus());
            }
        }
        String command = Command.SAVE
                .replace("{app}", app)
                .replace("{taskId}", String.valueOf(taskId))
                .replace("{taskLog}", JsonUtils.toJson(taskLog));
        RaftCommandResponse response = RaftCommandClientHolder.getRaftCommandClient().exec(command.getBytes(StandardCharsets.UTF_8));
        if (response.isSuccess()) return;
        throw new RuntimeException(JsonUtils.toJson(response));
    }
}
